﻿using Fast.Extensions.RabbitMQ;
using Fast.Reflection;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Polly;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System;
using System.Collections.Concurrent;
using System.Net.Sockets;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
using System.Timers;

namespace Fast.Extensions.EventBus.RabbitMQ
{
    /// <summary>
    /// 事件总线
    /// </summary>
    public class FastEventBusRabbitMQ : IEventBus, ISingletonDependency
    {
        protected IConnectionPool ConnectionPool { get; }
        protected FastRabbitMqEventBusOptions Options { get; }
        protected ILogger<FastEventBusRabbitMQ> Logger { get; }
        protected IEventBusSubscriptionsManager SubscriptionsManager { get; }

        /// <summary>
        /// 定时器，定时释放IModel
        /// </summary>
        private readonly Timer _timer = new Timer(1000);
        private ConcurrentBag<IModel> _waitDisposeChannel = new ConcurrentBag<IModel>();

        public FastEventBusRabbitMQ(IConnectionPool connectionPool, IOptions<FastRabbitMqEventBusOptions> options,
            IEventBusSubscriptionsManager subscriptionsManager, ILogger<FastEventBusRabbitMQ> logger)
        {
            ConnectionPool = connectionPool ?? throw new ArgumentNullException(nameof(connectionPool));
            Options = options?.Value ?? throw new ArgumentNullException(nameof(options));
            SubscriptionsManager = subscriptionsManager ?? new InMemoryEventBusSubscriptionsManager();
            Logger = logger ?? throw new ArgumentNullException(nameof(logger));

            _timer.Elapsed += new ElapsedEventHandler(DisposeChannel);
            //设置是执行一次（false）还是一直执行(true)；
            _timer.AutoReset = true;
            //是否执行Timer.Elapsed事件；
            _timer.Enabled = true;
            //启动定时器
            _timer.Start();
        }

        /// <summary>
        /// 发布事件
        /// </summary>
        /// <param name="event"></param>
        /// <returns></returns>
        public Task PublishAsync(IntegrationEvent @event)
        {
            var connection = ConnectionPool.Get(Options.ConnectionName);
            if (!connection.IsConnected)
            {
                connection.TryConnect();
            }

            var policy = Policy.Handle<BrokerUnreachableException>()
                .Or<SocketException>()
                .WaitAndRetry(Options.PublishFailRetryCount.Value,
                    retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) =>
                    {
                        Logger.LogWarning(ex, "在 {Timeout} 秒后无法发布事件: {EventId} ({ExceptionMessage})",
                            $"{time.TotalSeconds:n1}", @event.Id, ex.Message);
                    });
            var eventName = @event.GetType().Name;

            Logger.LogTrace("创建 RabbitMQ Channel 以发布事件: {EventId} ({EventName})", @event.Id,
                eventName);
            var channel = connection.CreateModel();
            //开启消息确认模式
            channel.ConfirmSelect();

            Logger.LogTrace("声明RabbitMQ Exchange 以发布事件: {EventId}", @event.Id);

            channel.ExchangeDeclare(Options.ExchangeName, "direct", true);

            var body = JsonSerializer.SerializeToUtf8Bytes(@event, @event.GetType(), new JsonSerializerOptions
            {
                WriteIndented = true
            });

            policy.Execute(() =>
            {
                var properties = channel.CreateBasicProperties();
                properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent;

                //1.监听我们一些不可达的消息。比如某些情况下交换机没有绑定到队列的情况下
                channel.BasicReturn += async (o, basic) =>
                {
                    _waitDisposeChannel.Add(channel);
                    await HandleBasicReturn(o, basic);
                };

                //2.消息发送成功的时候进入到这个事件。即RabbitMq服务器告诉生产者，我已经成功收到了消息
                channel.BasicAcks += async (o, basic) =>
                {
                    _waitDisposeChannel.Add(channel);
                    await HandleAck(o, basic);
                };

                //3.消息发送失败的时候进入到这个事件。即RabbitMq服务器告诉生产者，你发送的这条消息我没有成功的投递到Queue中，或者说我没有收到这条消息。
                channel.BasicNacks += async (o, basic) =>
                {
                    _waitDisposeChannel.Add(channel);
                    await HandleNack(o, basic);
                };

                Logger.LogTrace("发布事件到 RabbitMQ: {EventId}", @event.Id);

                channel.BasicPublish(
                    exchange: Options.ExchangeName,
                    routingKey: eventName,
                    mandatory: true,
                    basicProperties: properties,
                    body: body);
            });

            return Task.CompletedTask;
        }

        /// <summary>
        /// 处理不可达的消息后
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="basic"></param>
        /// <returns></returns>
        protected virtual Task HandleBasicReturn(object sender, BasicReturnEventArgs basic)
        {
            //消息失败的code
            var rc = basic.ReplyCode;
            //描述返回原因的文本。
            var rt = basic.ReplyText;
            //失败消息的内容
            var message = Encoding.UTF8.GetString(basic.Body.Span);

            Logger.LogInformation($"时间:{DateTime.Now};\r\npublish时调用了BasicReturn;\r\nMessage:{message};\r\nReplyCode:{rc};\r\nReplyText:{rt}");

            return Task.CompletedTask;
        }

        /// <summary>
        /// 处理发送成功后的消息
        /// </summary>
        /// <param name="o"></param>
        /// <param name="basic"></param>
        /// <returns></returns>
        protected virtual Task HandleAck(object o, BasicAckEventArgs basic)
        {
            Logger.LogInformation($"时间:{DateTime.Now};\r\npublish时调用了BasicAcks;\r\nDeliveryTag:{basic.DeliveryTag};\r\nMultiple:{basic.Multiple}");

            return Task.CompletedTask;
        }

        /// <summary>
        /// 处理发送失败后的消息
        /// </summary>
        /// <param name="o"></param>
        /// <param name="basic"></param>
        /// <returns></returns>
        protected virtual Task HandleNack(object o, BasicNackEventArgs basic)
        {
            //MQ服务器出现了异常，可能会出现Nack的情况
            Logger.LogInformation($"时间:{DateTime.Now};\r\npublish时调用了BasicNacks;\r\nDeliveryTag:{basic.DeliveryTag};\r\nMultiple:{basic.Multiple}");

            return Task.CompletedTask;
        }

        /// <summary>
        /// 订阅事件
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <typeparam name="TH"></typeparam>
        public async Task Subscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T>
        {
            var eventName = SubscriptionsManager.GetEventKey<T>();

            Logger.LogInformation("使用 {EventHandler} 订阅事件 {EventName} ", typeof(TH).GetGenericTypeName(), eventName);

            await DoInternalSubscription(eventName, MessageType.Normal);

            SubscriptionsManager.AddSubscription<T, TH>();
        }

        /// <summary>
        /// 订阅异常事件
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <typeparam name="TH"></typeparam>
        public async Task SubscribeException<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventExceptionHandler<T>
        {
            var eventName = SubscriptionsManager.GetEventKey<T>();

            Logger.LogInformation("使用 {EventHandler} 订阅事件 {EventName} ", typeof(TH).GetGenericTypeName(), eventName);

            await DoInternalSubscription(eventName, MessageType.Exception);

            SubscriptionsManager.AddSubscriptionException<T, TH>();
        }

        /// <summary>
        /// 处理事件
        /// </summary>
        /// <param name="ea"></param>
        /// <returns></returns>
        private async Task ProcessEventAsync(BasicDeliverEventArgs ea)
        {
            var eventName = ea.RoutingKey;
            var message = Encoding.UTF8.GetString(ea.Body.Span);

            Logger.LogTrace("处理 RabbitMQ 事件: {EventName}", eventName);

            if (SubscriptionsManager.HasSubscriptionsForEvent(eventName))
            {
                using (var scope = App.ServicesProvider.CreateScope())
                {
                    var handlerTypes = SubscriptionsManager.GetHandlersForEvent(eventName);
                    foreach (var handlerType in handlerTypes)
                    {
                        var handler = scope.ServiceProvider.GetService(handlerType);
                        if (handler == null) continue;
                        var eventType = SubscriptionsManager.GetEventTypeByName(eventName);
                        var integrationEvent = JsonSerializer.Deserialize(message, eventType, new JsonSerializerOptions() { PropertyNameCaseInsensitive = true });

                        Type concreteType;
                        if (typeof(IIntegrationEventHandler).IsAssignableFrom(handler.GetType()))
                            concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
                        else
                            concreteType = typeof(IIntegrationEventExceptionHandler<>).MakeGenericType(eventType);
                        await Task.Yield();
                        await (Task)concreteType.GetMethod("HandleAsync").Invoke(handler, new object[] { integrationEvent });
                    }
                }
            }
            else
            {
                Logger.LogWarning("没有订阅 RabbitMQ 事件: {EventName}", eventName);
            }
        }

        /// <summary>
        /// 订阅
        /// </summary>
        /// <param name="eventName"></param>
        /// <returns></returns>
        private Task DoInternalSubscription(string eventName, MessageType messageType)
        {
            var containsKey = SubscriptionsManager.HasSubscriptionsForEvent(eventName);
            if (!containsKey)
            {
                var connection = ConnectionPool.Get(Options.ConnectionName);
                if (!connection.IsConnected)
                {
                    connection.TryConnect();
                }

                var consumer = App.ServicesProvider.GetService<IRabbitMqMessageConsumer>();

                consumer.OnMessageReceived(ProcessEventAsync);

                consumer.StartBasicConsume(new BasicConsumeContext(
                    new ExchangeDeclareConfiguration(
                       Options.ExchangeName,
                       type: "direct",
                       durable: true
                    ),
                    new QueueDeclareConfiguration(
                        $"{AssemblyHelper.GetEntryAssemblyName().ToLower()}.{eventName.ToLower()}",
                        durable: true,
                        exclusive: false,
                        autoDelete: false
                    ),
                    Options.ConnectionName, eventName, Options.EnableDeadLetterExchange.Value, messageType
                ));
            }
            return Task.CompletedTask;
        }

        /// <summary>
        /// 释放发布消息生成的IModel对象
        /// </summary>
        /// <param name="source"></param>
        /// <param name="e"></param>
        private void DisposeChannel(object source, ElapsedEventArgs e)
        {
            _timer.Stop();
            while (_waitDisposeChannel.Count > 0)
            {
                try
                {
                    _waitDisposeChannel.TryTake(out var channel);
                    channel.Dispose();
                }
                catch (Exception ex)
                {
                    Logger.LogWarning(ex.ToString());
                }
            }
            _timer.Start();
        }
    }
}